package com.haiyou.data.common.persister.mysql.dao;

import java.sql.*;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

import com.haiyou.common.jdbc.DbType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.haiyou.common.system.Properties;
import com.haiyou.common.system.SystemProperties;
import com.haiyou.common.thread.task.Slowlable;
import com.haiyou.common.util.mysql.BatchUtils;
import com.haiyou.data.common.persister.mysql.help.DatabaseHelper;

public interface BaseDao extends Slowlable {

	Logger log = LoggerFactory.getLogger(BaseDao.class);

	@Override
	default int readSlowly() {
		return Properties.slowly_mysqlRead();
	}

	@Override
	default int writeSlowly() {
		return Properties.slowly_mysqlWrite();
	}

	@Override
	default int executeSlowly() {
		return Properties.slowly_mysqlExecute();
	}

	/** 回收连接 */
	default void off(ResultSet resultSet, Statement statement, Connection connection) {
		DatabaseHelper.closeConnection(resultSet, statement, connection);
	}

	/** 打开连接 */
	default Connection on(String db) {
		return DatabaseHelper.openConnection(db);
	}

	/** 执行一条语句 */
	default boolean execute(String db, String sql) {
		return execute(on(db), sql);
	}

	/** 执行一条语句 */
	default boolean execute(Connection connection, String sql) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		PreparedStatement preparedStatement = null;
		try {
			preparedStatement = connection.prepareStatement(sql);
			return preparedStatement.execute();
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(null, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly()) {
				log.warn("SLOWLY: [{}] elapsed [{} ms]", sql, elapsed);
			} else {
				if (log.isDebugEnabled()) {
					log.debug("[{}] elapsed [{} ms]", sql, elapsed);
				}
			}
		}
	}

	/** 执行一条查询语句，并自定义读取规则,注意：如果成员有char类型的，设置的长度和实际数据长度并不一致的，需要调用trim()函数 */
	default void executeQuery(String db, String sql, Consumer<ResultSet> consumer) {
		executeQuery(on(db), sql, consumer);
	}

	default boolean checkTableIsExist(String table) {
		String sql = String.format("show tables like '%s' ", "%" + table + "%");
		AtomicReference<String> tab = new AtomicReference<>("");
		BaseDaoImpl.getInstance().executeQuery(DbType.LOG_DB, sql, rs -> {
			try {
				tab.set(rs.getString(1));
			} catch (SQLException e) {
				throw new RuntimeException(e);
			}
		});
		return tab.get().equals(table);
	}

	/**
	 * 统计任务使用
	 */
	default void executeQuery(String db, String sql, String tableName,Consumer<ResultSet> consumer) {
		// 检查表是否存在
		if (!checkTableIsExist(tableName)) {
			log.error("数据表{}，不存在", tableName);
			return;
		}
		executeQuery(on(db), sql, consumer);
	}

	/** 执行一条查询语句，并自定义读取规则,注意：如果成员有char类型的，设置的长度和实际数据长度并不一致的，需要调用trim()函数 */
	default void executeQuery(Connection connection, String sql, Consumer<ResultSet> consumer) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		PreparedStatement preparedStatement = null;
		ResultSet resultSet = null;
		try {
			preparedStatement = connection.prepareStatement(sql);
			resultSet = preparedStatement.executeQuery();
			while (resultSet != null && resultSet.next())
				consumer.accept(resultSet);
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(resultSet, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > readSlowly()) {
				log.warn("SLOWLY: [{}] elapsed [{} ms]", sql, elapsed);
			} else {
				if (log.isDebugEnabled()) {
					log.debug("[{}] elapsed [{} ms]", sql, elapsed);

				}
			}
		}
	}

	/** 执行一条查询语句并组装成Map结果集返回 */
	default List<Map<String, Object>> executeQuery(String db, String sql) {
		return executeQuery(on(db), sql);
	}

	/** 执行一条查询语句并组装成Map结果集返回 */
	default List<Map<String, Object>> executeQuery(Connection connection, String sql) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		PreparedStatement preparedStatement = null;
		ResultSet resultSet = null;
		List<Map<String, Object>> list = Lists.newArrayList();
		try {
			preparedStatement = connection.prepareStatement(sql);
			resultSet = preparedStatement.executeQuery();
			if (resultSet != null) {
				ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
				while (resultSet != null && resultSet.next()) {
					Map<String, Object> map = Maps.newLinkedHashMap();
					for (int i = 0; ++i <= resultSetMetaData.getColumnCount();) {
						String name = resultSetMetaData.getColumnLabel(i);
						Object value = resultSet.getObject(name);
						map.put(name, value);
					}
					list.add(map);
				}
			}
			return list;
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(resultSet, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > readSlowly())
				log.warn("SLOWLY: [{}] elapsed [{} ms,size={}]", sql, elapsed, list.size());
			else
				log.debug("[{}] elapsed [{} ms,size={}]", sql, elapsed, list.size());
		}
	}

	default int sqlBatchNum() {
		return Properties.sql_batchNum();
	}

	/** 批量执行语句 */
	default int[] executeBatch(String db, Iterable<String> sqls) {
		int size = Iterables.size(sqls);
		if (size <= sqlBatchNum())
			return executeBatch(on(db), sqls);
		else {
			List<List<String>> lists = BatchUtils.toLists(sqls, sqlBatchNum());
			int[] result = null;
			for (List<String> list : lists) {
				int[] r = executeBatch(on(db), list);
				if (result == null) {
					result = r;
				} else {
					int[] rs = new int[result.length + r.length];
					System.arraycopy(result, 0, rs, 0, result.length);
					System.arraycopy(r, 0, rs, result.length, r.length);
					result = rs;
				}
			}
			return result;
		}
	}

	/** 批量执行语句 */
	default int[] executeBatch(Connection connection, Iterable<String> sqls) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		Statement statement = null;
		try {
			int size = Iterables.size(sqls);
			if (size > sqlBatchNum()) {
				log.warn("sql  statement too more!{}", size);
			}
			statement = connection.createStatement();
			Iterator<String> iterator = sqls.iterator();
			while (iterator.hasNext()) {
				String sql = iterator.next();
				statement.addBatch(sql);
			}
			return statement.executeBatch();
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sqls, e);
		} finally {
			off(null, statement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly()) {
				log.warn("SLOWLY: batch[{}] elapsed [{} ms]", Iterables.size(sqls), elapsed);
			} else {
				if (log.isDebugEnabled()) {
					log.debug("batch[{}] elapsed [{} ms]", Iterables.size(sqls), elapsed);
				}
			}
		}
	}

	/** 执行更新语句 */
	default int executeUpdate(Connection connection, String sql) {
		Stopwatch stopwatch = Stopwatch.createStarted();
		PreparedStatement preparedStatement = null;
		try {
			preparedStatement = connection.prepareStatement(sql);
			return preparedStatement.executeUpdate();
		} catch (Exception e) {
			throw new RuntimeException(e.getMessage() + ":" + sql, e);
		} finally {
			off(null, preparedStatement, connection);
			long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
			if (elapsed > executeSlowly()) {
				log.warn("SLOWLY: [{}] elapsed [{} ms]", sql, elapsed);
			} else {
				if (log.isDebugEnabled()) {
					log.debug("[{}] elapsed [{} ms]", sql, elapsed);
				}
			}

		}
	}

	/** 执行更新语句 */
	default int executeUpdate(String db, String sql) {
		return executeUpdate(on(db), sql);
	}

}
